package com.atguigu.day07;

import com.atguigu.bean.Event;
import com.atguigu.day03.Flink05_Source_Custom;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class Flink10_KeyedState_MapState {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        env
                .addSource(new Flink05_Source_Custom.ClickSource())
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {

                            @Override
                            public long extractTimestamp(Event event, long l) {
                                return event.timestamp;
                            }
                        }))
                .keyBy(r -> r.url)
                //我们要计算的是每一个url在每一个窗口中的pv数据
                .process(new MyTumblingWindow(Time.seconds(5)))
                .print();
        env.execute();

    }

    public static class MyTumblingWindow extends KeyedProcessFunction<String, Event, String>{
        //1.定义状态 K:窗口的开始时间  v：属于这个窗口开始时间的pv个数
        private MapState<Long, Tuple2<String,Integer>> mapState;

        //窗口大小
        private Long windowSize;

        public MyTumblingWindow(Time size) {
            windowSize = size.toMilliseconds();
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            mapState = getRuntimeContext().getMapState(new MapStateDescriptor<Long, Tuple2<String,Integer>>("map-state", Types.LONG,Types.TUPLE(Types.STRING,Types.INT)));
        }
        @Override
        public void processElement(Event value, KeyedProcessFunction<String, Event, String>.Context ctx, Collector<String> out) throws Exception {
            //1.获取窗口的开始时间 只要属于同一个窗口那么开始时间都是一样的
            long windowStart = ctx.timestamp() - (ctx.timestamp() + windowSize) % windowSize;

            //获取窗口的结束时间
            long windowEnd = windowStart + windowSize;

            //2.根据窗口的开始时间获取当前这个窗口的url的pv个数
            if (mapState.contains(windowStart)){
                Tuple2<String, Integer> tuple2 = mapState.get(windowStart);
                tuple2.f1 = tuple2.f1 + 1;
                mapState.put(windowStart,tuple2);
            }else {
                mapState.put(windowStart,Tuple2.of(value.url, 1));
            }

            //注册一个定时器，定时时间为窗口最大时间戳 也就是 windowEnd-1ms
            ctx.timerService().registerEventTimeTimer(windowEnd-1);
        }

        @Override
        public void onTimer(long timestamp, KeyedProcessFunction<String, Event, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
            long windowEnd = timestamp + 1;
            long windowStart = windowEnd - windowSize;
            out.collect("窗口：["+windowStart+","+windowEnd+"]"+"pv:"+mapState.get(windowStart));

            //关闭窗口
            mapState.remove(windowStart);
        }
    }
}
